package spark.core.rdd

import org.apache.spark.rdd.RDD
import org.apache.spark.{SparkConf, SparkContext}

object rddLineage {

  def main(args: Array[String]): Unit = {
    // Todo 1. RDD 血缘关系
    // RDD只支持粗粒度转换，即在大量记录上执行的单个操作。将创建RDD的一系列Lineage（血统）记录下来，以便恢复丢失的分区。
    // RDD的Lineage会记录RDD的元数据信息和转换行为.
    // 当该RDD的部分分区数据丢失时，它可以根据这些信息来重新运算和恢复丢失的数据分区。

    // Todo 2. RDD 依赖关系
    // 这里所谓的依赖关系，其实就是两个相邻RDD之间的关系

    // Todo 3. RDD 窄依赖
    // 窄依赖表示每一个父(上游)RDD的Partition 最多 被子（下游）RDD的一个Partition使用，窄依赖我们形象的比喻为独生子女。
    // class OneToOneDependency[T](rdd: RDD[T]) extends NarrowDependency[T](rdd)
    // Todo 3. RDD 宽依赖
    // 宽依赖表示同一个父（上游）RDD的Partition被多个子（下游）RDD的Partition依赖，会引起Shuffle.
    // 总结：宽依赖我们形象的比喻为多生。
    // class ShuffleDependency[K: ClassTag, V: ClassTag, C: ClassTag](...) extends Dependency[Product2[K, V]]

    // 1.创建SparkConf并设置App名称
    val sparkConf: SparkConf = new SparkConf().setMaster("local[*]").setAppName("rddSerializable")
    // 2. 创建SparkContext，该对象是提交Spark App的入口
    val sparkContext: SparkContext = new SparkContext(sparkConf)

    val fileRDD: RDD[String] = sparkContext.textFile("Input/test.txt")
    // RDD 血缘关系
    println(fileRDD.toDebugString)
    // RDD 依赖关系
    println(fileRDD.dependencies)
    println("----------------------")

    val wordRDD: RDD[String] = fileRDD.flatMap(_.split(" "))
    println(wordRDD.toDebugString)
    println(wordRDD.dependencies)
    println("----------------------")

    val mapRDD: RDD[(String, Int)] = wordRDD.map((_, 1))
    println(mapRDD.toDebugString)
    println(mapRDD.dependencies)
    println("----------------------")

    val resultRDD: RDD[(String, Int)] = mapRDD.reduceByKey(_ + _)
    println(resultRDD.toDebugString)
    println(resultRDD.dependencies)

    resultRDD.collect()

    sparkContext.stop()
  }

}
